package com.neuedu.music;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.TreeMap;

public class MusicReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private static final Text MUSIC_LIST = new Text();
    private static final IntWritable PLAY_COUNT = new IntWritable();
    private static TreeMap<Integer, String> treemap;

    @Override
    protected void setup(Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //初始化事件:集合实例化
        treemap = new TreeMap<Integer, String>();
    }

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //接收的map数据格式
        //<歌曲名称，<1,1,1,1,1>>
        //统计歌曲的播放次数
        int sum = 0;
        for (IntWritable v : values) {
            sum += v.get();
        }
        //存储数据至集合中
        String music_name = key.toString();
        //若键存在时，覆盖之，所以需要判断，也就是允许歌曲并列
        if (treemap.containsKey(sum)) {
            //存在则并列
            String value = treemap.get(sum) + "," + music_name;
            treemap.put(sum, value);
        } else {
            //播放次数不存在，直接添加
            treemap.put(sum, music_name);
        }
    }

    @Override
    protected void cleanup(Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //在reducer任务结束时调用一次
        //将集合中的数据从高到低，依次输出
        for (Integer key : treemap.descendingKeySet()) {
            //输出
            MUSIC_LIST.set(treemap.get(key));
            PLAY_COUNT.set(key);
            context.write(MUSIC_LIST, PLAY_COUNT);
        }
    }
}